package com.wolf.etl.core;

import com.alibaba.druid.support.json.JSONUtils;
import com.wolf.etl.core.database.DynamicService;
import com.wolf.etl.core.filter.AbstractFilter;
import com.wolf.etl.core.rule.check.AbstractCheckRule;
import com.wolf.etl.core.rule.convert.AbstractConvertRule;
import com.wolf.etl.core.task.TaskStatus;
import com.wolf.etl.core.transform.AbstractTransform;
import com.wolf.etl.enumeration.*;
import com.wolf.etl.model.*;
import com.wolf.etl.service.*;
import com.wolf.etl.vo.ColumnDesc;
import com.wolf.etl.vo.Database;
import com.wolf.etl.vo.Table;
import com.wolf.exception.ETL_ERROR_MSG;
import com.wolf.exception.WolfException;
import com.wolf.provider.order.manager.BeanManager;
import com.wolf.vo.BaseResult;
import com.wolf.vo.KeyValuePair;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.sql.DataSource;
import java.sql.SQLException;
import java.util.*;

/**
 * Created by tommy's father on 2017/8/18.
 */
@Service("eTLManager")
public class ETLManager {

    private static Logger logger = LoggerFactory.getLogger(ETLManager.class);

    private static final String logger_prefix = "wolf-etl#com.wolf.etl.core#ETLManager";

    @Autowired
    private BeanManager beanManager;

    @Autowired
    private InitService initService;

    @Autowired
    private IEtlDefineService etlDefineService;

    @Autowired
    private IEtlTableService etlTableService;

    @Autowired
    private IEtlTableRelService relService;

    @Autowired
    private IEtlTransformDefineService transformDefineService;

    @Autowired
    private IEtlTableFilterService filterService;

    @Autowired
    private IEtlTransformRuleService ruleService;

    @Autowired
    private IEtlTransformRuleTypeService ruleTypeService;

    @Autowired
    private IEtlDataSourceService dataSourceDefineService;

    @Autowired
    private IEtlTableOrderbyService etlTableOrderbyService;

    @Autowired
    private DynamicService dynamicService;

    @Autowired
    private IEtlErrorLogService etlErrorLogService;

    @Autowired
    private IEtlTransformFieldTypeService etlTransformFieldTypeService;

    @Autowired
    private IEtlTransformRuleService etlTransformRuleService;

    @Autowired
    private IEtlTransformRuleTypeService etlTransformRuleTypeService;

    @Autowired
    private IEtlFunctionService etlFunctionService;

    @Autowired
    private IEtlTableRelRuleService etlTableRelRuleService;

    @Autowired
    private IEtlRewriteBackService etlRewriteBackService;

    @Autowired
    private IEtlInstanceService etlInstanceService;

    @Autowired
    private IEtlTaskService etlTaskService;

    @Autowired
    private IEtlHandlerService etlHandlerService;

    @Autowired
    private IEtlJavascriptParamService javascriptParamService;

    @Autowired
    private IEtlJavascriptService javascriptService;

    @Autowired
    private IEtlApiService etlApiService;


    /**
     * 试运行标识
     */
    private Boolean isTest = false;

    private static final String SELECT_TABLE_JOIN = "select %s from %s where 1=1 %s %s {#limit#} %s";
    private static final String PREFFIX_FIELD = " %s.%s ";
    private static final String TABLE_ALIAS = " %s %s ";
    private static final String TABLE_JOIN_STR = " and %s=%s ";

    private static final String SELECT_PART_STR = "%s.*";
    public static final String COMMA = ",";
    private static final String ASTERISK = "*";
    private static final String REPLACE_HOLDER = "%s";
    private static final String SQL_STR_VALUE = "'%s'";
    private static final String ORDER_BY_STR = " order by ";
    private static final String FILTER_STR = " and %s.%s=%s ";

    public static final String MAIN_DETAIL_REFERENCE_KEY = "MAIN_DETAIL_REFERENCE_KEY";

    private EtlTableModel srcTableModel;
    private boolean isChildrens = false;
    private int[] results = null;
    private boolean needCommit = false;
    private List<String> childrenFilters = new ArrayList<String>();
    private StringBuffer selects = new StringBuffer();
    private StringBuffer tables = new StringBuffer();
    private StringBuffer joins = new StringBuffer();
    private StringBuffer filters = new StringBuffer();
    private Long defind_id;
    private Long tabId = 0L;
    private Long pTabId = 0L;
    private Long srcDataSourceId;
    private Long targetDataSourceId;
    private List<Long> tabs = new ArrayList<Long>();
    private List<String> fields = new ArrayList<String>();
    private List<String> values = new ArrayList<String>();
    private List<KeyValuePair> orderbys = new ArrayList<KeyValuePair>();
    private AbstractTransform transform;
    private KeyValuePair referenceKeyValuePair = new KeyValuePair();
    private DatabaseManager srcDatabaseManager;
    private DatabaseManager tarDatabaseManager;
    private EtlDefineModel etlDefineModel = null;
    private EtlInstanceModel instance;

    private List<String> sourceSqls = new ArrayList<String>();
    private List<String> targetSqls = new ArrayList<String>();
    private List<String> writeBackSqls = new ArrayList<String>();

    private Map<Long, Database> databaseMap = new HashMap<Long, Database>();

    private Map<Long, EtlTransformRuleTypeModel> ruleTypeMap = new HashMap<Long, EtlTransformRuleTypeModel>();

    public void setReferenceKey(KeyValuePair keyValuePair) {
        this.referenceKeyValuePair = keyValuePair;
    }

    public void needCommit() {
        needCommit = true;
    }

    public void setResults(int[] results) {
        this.results = results;
    }

    public Long getTabId() {
        if (srcTableModel != null) {
            tabId = srcTableModel.getId();
        }
        return tabId;
    }

    public Long getDefindId() {
        if (etlDefineModel != null) {
            return etlDefineModel.getId();
        }
        return defind_id;
    }

    public BeanManager getBeanManager() {
        return beanManager;
    }

    public void setBeanManager(BeanManager beanManager) {
        this.beanManager = beanManager;
    }

    public Long getParentTabId() {
        return pTabId;
    }

    public Long getSrcDataSourceId() {
        return srcDataSourceId;
    }

    public Long getTargetDataSourceId() {
        return targetDataSourceId;
    }

    public List<String> getFields() {
        return fields;
    }

    public List<String> getValues() {
        return values;
    }

    public EtlTableModel getTableModel() {
        return srcTableModel;
    }

    public EtlDefineModel getEtlDefineModel() {
        return etlDefineModel;
    }

    private boolean isNumberType(Long tabId, String columnName) {
        return false;
    }

    public List<EtlTableFilterModel> findEtlTableFilterModels(Long tabId) {
        return filterService.findEtlFilterModels(tabId);
    }

    private List<EtlTransformDefineModel> findByDefineId(Long etl_define_id) {
        return transformDefineService.findByDefineId(etl_define_id);
    }

    private EtlTableRelModel findEtlTableExtRelModel(Long masterTabId, Long relTabId) {
        return relService.findEtlTableExtRelModel(masterTabId, relTabId);
    }

    public List<EtlTableRelModel> findEtlTableSlaveRelModels(Long masterTabId) {
        return relService.findEtlTableSlaveRelModels(masterTabId);
    }

    public List<EtlTransformFieldTypeModel> findByDialect(DIALECT dialect) {
        return etlTransformFieldTypeService.findByDialect(dialect.getValue());
    }

    public List<EtlTransformRuleModel> findTransRules() {
        return etlTransformRuleService.findRuleByDefindId(defind_id);
    }

    public List<EtlTableRelRuleModel> findTableRelRules() {
        return etlTableRelRuleService.findRuleByDefindId(defind_id);
    }

    public EtlTableRelRuleModel findTableRelRuleById(Long id) {
        return etlTableRelRuleService.findById(id);
    }

    public List<EtlRewriteBackModel> findRewriteBack() {
        return etlRewriteBackService.findRewriteByDefindId(defind_id);
    }

    public EtlInstanceModel findInstanceById(Long instance_id) {
        return etlInstanceService.findById(instance_id);
    }

    public List<EtlInstanceModel> findInstancesByTaskId(Long taskId) {
        return etlInstanceService.findByTaskId(taskId);
    }

    public void updateInstance(EtlInstanceModel model) {
        etlInstanceService.update(model);
    }

    public void insertInstance(EtlInstanceModel instance) {
        etlInstanceService.insertModel(instance);
    }

    public EtlInstanceModel findInstance(Long task_id, String instance_code) {
        EtlInstanceModel model = etlInstanceService.findInstance(task_id, instance_code);
        return model;
    }

    public EtlJavascriptModel findJavascriptById(Long id) {
        return javascriptService.findById(id);
    }

    public EtlApiModel findApiById(Long id) {
        return etlApiService.findById(id);
    }

    public List<EtlJavascriptParamModel> findJavascriptParams(Long functionId) {
        return javascriptParamService.findByFunctionId(functionId);
    }

    public EtlTaskModel findTaskById(Long taskId) {
        return etlTaskService.findById(taskId);
    }

    public List<EtlInstanceModel> findInstancesByDefineId(Long define_id) {
        return etlInstanceService.findByDefineId(define_id);
    }

    public Map<Long, EtlTransformRuleTypeModel> findRuleTypeMap() {
        if (!CollectionUtils.isEmpty(ruleTypeMap)) {
            return ruleTypeMap;
        }
        List<EtlTransformRuleTypeModel> ruleTypeList = etlTransformRuleTypeService.findAll();
        for (EtlTransformRuleTypeModel type : ruleTypeList) {
            ruleTypeMap.put(type.getId(), type);
        }
        return ruleTypeMap;
    }

    public List<EtlHandlerModel> findHandlerList() {
        return etlHandlerService.findByDefineId(defind_id);
    }

    public EtlTableRelModel findEtlTableSlaveRelModel(Long masterTabId, Long slaveId) {
        List<EtlTableRelModel> models = findEtlTableSlaveRelModels(masterTabId);
        for (EtlTableRelModel model : models) {
            if (model.getSlave_table_id() == slaveId) {
                return model;
            }
        }
        return null;
    }

    public List<EtlTableModel> findMasterEtlTableModels(Long defId) {
        return etlTableService.findMasterEtlTableModels(defId);
    }

    public EtlFunctionModel findFunctionById(Long id) {
        return etlFunctionService.findById(id);
    }

    private boolean testMasterTableFilter(Long tabId) {
        return filterService.testMasterTableFilter(tabId);
    }

    public List<EtlTaskModel> findTaskByStatus(TaskStatus status) {
        List<EtlTaskModel> vos = etlTaskService.findByStatus(status.toString());
        return vos;
    }

    public List<EtlInstanceModel> findAllInstance() {
        return etlInstanceService.findAll();
    }

    private Map<KeyValuePair, EtlTableFilterModel> buildCondition(Long tabId) {
        Map<KeyValuePair, EtlTableFilterModel> results = new HashMap<KeyValuePair, EtlTableFilterModel>();
        List<EtlTableFilterModel> infos = findEtlTableFilterModels(tabId);
        for (EtlTableFilterModel info : infos) {
            KeyValuePair kvp = new KeyValuePair();

            Long masterTabId = info.getMaster_table_id();
            if (masterTabId > 0) {

                EtlTableModel model = findEtlTableModel(masterTabId);
                String alias = getTableAlias(model);

                kvp.setKey(String.valueOf(masterTabId));
                kvp.setValue(alias);

                results.put(kvp, info);
                continue;
            }

//            Long etlTabId = info.getEtl_table_id();
            EtlTableModel model = findEtlTableModel(masterTabId);
            String alias = getTableAlias(model);
            kvp.setKey(String.valueOf(masterTabId));
            kvp.setValue(alias);

            results.put(kvp, info);
        }

        return results;
    }

    public EtlTableModel findEtlTableModel(Long tabId) {
        EtlTableModel model = etlTableService.findById(tabId);
        if (model == null) {
            throw new WolfException(String.format("Can't find any etl table info by id[tab id=>%s].", tabId));
        }
        model.check();
        return model;
    }

    public List<EtlTableOrderbyModel> findOrderby(Long tableId) {
        List<EtlTableOrderbyModel> vos = etlTableOrderbyService.findByTableId(tableId);
        return vos;
    }

    public EtlDefineModel findDefineById(Long id) {
        return etlDefineService.findById(id);
    }

    private List<EtlDefineModel> findAllEtlDefineModels() {
        return etlDefineService.findAllEtlDefineModels();
    }

    private List<EtlTableModel> findAllEtlExtTableModels(Long tabId) {
        return etlTableService.findAllExtTableModelsById(tabId);
    }

    private void testExistsTable(Long tableId, String columnName) {
        if (!tabs.contains(tableId)) {
            tabs.add(tableId);
            EtlTableModel etlTableModel = findEtlTableModel(tableId);
            Database database = getDataBaseInstance(etlTableModel.getData_source_id());
            database = getTable(etlTableModel.getData_source_id(), database, etlTableModel.getName());
            Table table = database.getTable(etlTableModel.getName());
            if (table == null) {
                throw new WolfException(String.format("Datasource[id=>%s] can't find table[id=>%s,name=>%s]",
                        etlTableModel.getData_source_id(),
                        etlTableModel.getId(),
                        etlTableModel.getName()));
            }

            if (table.getColumn(columnName) == null) {
                throw new WolfException(String.format("Datasource[id=>%s] can't find any column info[table id=>%s,table name=>%s,column name=>%s]",
                        etlTableModel.getData_source_id(),
                        etlTableModel.getId(),
                        etlTableModel.getName(),
                        columnName));
            }
        }
    }

    private void testExistsTable(Long tableId) {
        if (!tabs.contains(tableId)) {
            tabs.add(tableId);
            EtlTableModel etlTableModel = findEtlTableModel(tableId);
            Database database = getDataBaseInstance(etlTableModel.getData_source_id());
            database = getTable(etlTableModel.getData_source_id(), database, etlTableModel.getName());
            Table table = database.getTable(etlTableModel.getName());
            if (table == null) {
                throw new WolfException(String.format("Datasource[id=>%s] can't find table[id=>%s,name=>%s]",
                        etlTableModel.getData_source_id(),
                        etlTableModel.getId(),
                        etlTableModel.getName()));
            }
        }
    }

    private void testExistsTable(EtlTransformDefineModel transformDefineModel) {
        try {
            Long srcTableId = transformDefineModel.getSrc_table_id();
            Long targetTableId = transformDefineModel.getTarget_table_id();
            String srcColumnName = transformDefineModel.getSrc_column_name();
            String targetColumnName = transformDefineModel.getTarget_column_name();
            testExistsTable(srcTableId, srcColumnName);
            testExistsTable(targetTableId, targetColumnName);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private boolean hasChildrens() {
        List<EtlTableRelModel> relModels = findEtlTableSlaveRelModels(tabId);
        if (CollectionUtils.isEmpty(relModels)) {
            return false;
        }
        return true;
    }

    private void executeDetail(int parentKeyId) {
        List<EtlTableRelModel> relModels = findEtlTableSlaveRelModels(tabId);
        if (CollectionUtils.isEmpty(relModels)) {
            return;
        }
        childrenFilters.clear();

        List<EtlTableModel> tableModels = new ArrayList<EtlTableModel>();
        for (EtlTableRelModel relModel : relModels) {
            EtlTableModel model = findEtlTableModel(relModel.getSlave_table_id());
            childrenFilters.add(String.format(FILTER_STR, model.getAlias(), relModel.getSlave_column(), parentKeyId));
            tableModels.add(model);
        }

        executeMain(tableModels);
    }

    private void executeDetail() {
        List<EtlTableRelModel> relModels = findEtlTableSlaveRelModels(tabId);
        if (CollectionUtils.isEmpty(relModels)) {
            return;
        }

        List<EtlTableModel> tableModels = new ArrayList<EtlTableModel>();
        for (EtlTableRelModel relModel : relModels) {
            EtlTableModel model = findEtlTableModel(relModel.getSlave_table_id());
            tableModels.add(model);
        }
        executeMain(tableModels);
    }

    private void appendRefenceKey() {
        if (isChildrens) {
            selects.append(String.format("'%s' %s,", referenceKeyValuePair.getValue(), MAIN_DETAIL_REFERENCE_KEY));
            isChildrens = false;
        }
    }

    private void executeMain(List<EtlTableModel> models) {
        for (EtlTableModel model : models) {
            reset();
            this.srcTableModel = model;
            appendRefenceKey();
            transform = AbstractTransform.getInstance(this);
            List<EtlTransformDefineModel> etlTransformDefineModels = findByDefineId(getDefindId());
            for (EtlTransformDefineModel transformDefineModel : etlTransformDefineModels) {
                testExistsTable(transformDefineModel);

//                transform.beforeTransform(transformDefineModel);
            }
            transform.beforeTransform(etlTransformDefineModels);
            BaseResult result = transform.executeTransform();

//            if (!result.getSuccess()) {
//                throw new WolfException("xxx");
//            }

//            doDetailTrans();
        }
    }

    private void doDetailTrans() {
        if (hasChildrens()) {
            if (IS_DYNAMIC.YES.isEquals(srcTableModel.getIs_dynamic())) {
                isChildrens = true;
                executeDetail();
                return;
            }

            for (int parentKeyId : results) {
                executeDetail(parentKeyId);
            }
        }
    }

    public void count(Long defindId) {

        EtlDefineModel etlDefineModel = findDefineById(defindId);
        if (etlDefineModel == null) {
            WolfException.report(ETL_ERROR_MSG.ERROR_TRANSFORM_RULE_009, String.format("找不到ETL定义，id=%s", defindId));
        }
        init(etlDefineModel);
        List<EtlTableModel> models = findMasterEtlTableModels(etlDefineModel.getId());
        for (EtlTableModel model : models) {
            reset();
            this.srcTableModel = model;
            appendRefenceKey();
            transform = AbstractTransform.getInstance(this);
            List<EtlTransformDefineModel> etlTransformDefineModels = findByDefineId(getDefindId());
            for (EtlTransformDefineModel transformDefineModel : etlTransformDefineModels) {
                testExistsTable(transformDefineModel);
            }
            transform.beforeTransform(etlTransformDefineModels);
            String sql = transform.getCountSql();

            List<Map<String, ColumnDesc>> result = srcDatabaseManager.queryResults(sql);
            if (CollectionUtils.isEmpty(result)) {
                return;
            }
            ColumnDesc record_count = result.get(0).get("record_count");
            if (record_count == null) {
                record_count = result.get(0).get("RECORD_COUNT");
            }
            Long count = Long.parseLong(record_count.getObj().toString());
            etlDefineModel.setRecord_count(count);
            etlDefineService.updateModel(etlDefineModel);
        }
    }

    public BaseResult execute(Long task_id) {
        setTest(false);

        String code = initService.getInstanceCode();
        instance = findInstance(task_id, code);

        // 没有实例
        if (instance == null) {
            return BaseResult.getFailure("task_id:%s instance code is null", task_id);
        }

        // 异常
        if (instance.getStatus() == INSTANCE_STATUS.ERROR.getValue()) {
            logger.info("任务异常");
            return BaseResult.getFailure("Task %s isn't enable", code);
        }

        // 完成
        if (instance.getStatus() == INSTANCE_STATUS.FINISH.getValue()) {
            logger.info("任务已完成");
            return BaseResult.getSuccess(String.format("Task %s is finished", code));
        }

        // 判断是否完成
        if (instance.getPage_num() == instance.getMax_page_num()) {
            instance.setStatus(INSTANCE_STATUS.FINISH.getValue());
            updateInstance(instance);
            logger.info("任务已完成");
            return BaseResult.getSuccess(String.format("Task %s is finished", code));
        }

        EtlTaskModel task = findTaskById(task_id);

        EtlDefineModel etlDefineModel = findDefineById(task.getEtl_define_id());
        if (etlDefineModel == null) {
            return BaseResult.getSuccess();
        }
        init(etlDefineModel);
        List<EtlTableModel> models = findMasterEtlTableModels(etlDefineModel.getId());
        executeMain(models);
        BaseResult result = doExecute();

        //commit();

        instance.addPageNum();
        etlInstanceService.update(instance);

        return result;
    }

    public Integer getNextPage() {
        if (instance == null) {
            return 0;
        }
        return instance.getPage_num();
    }

    /**
     * 试运行
     *
     * @param define_id
     */
    public void test(Long define_id) {

        setTest(true);

        EtlDefineModel defineModel = etlDefineService.findById(define_id);
        init(etlDefineModel);
        List<EtlTableModel> models = findMasterEtlTableModels(etlDefineModel.getId());
        executeMain(models);
        BaseResult result = doExecute();
        commit();
    }

    public void execute() {
        setTest(false);
        List<EtlDefineModel> etlDefineModels = findAllEtlDefineModels();
        for (EtlDefineModel etlDefineModel : etlDefineModels) {
            init(etlDefineModel);
            List<EtlTableModel> models = findMasterEtlTableModels(etlDefineModel.getId());
            executeMain(models);
            BaseResult result = doExecute();
            if (!result.isSucc()) {
                continue;
            }
            commit();
        }
    }

    private BaseResult doExecute() {

        BaseResult result = doSrcExecute();
        if (result.isFailure()) {
            return result;
        }

        // 非试运行，进行数据插入和回写操作
        if (!getTest()) {
            // 写入目标表数据
            result = doTarExecute();
            if (result.isFailure()) {
                return result;
            }

            // 回写源表字段信息
            result = doSrcWriteBack();
        }

        return result;
    }

    private void commit() {
        try {
            srcDatabaseManager.commit();
        } catch (SQLException e) {
            throw new WolfException(String.format("SQLException error:%s", e.getMessage()));
        }

//        if (!needCommit) {
//            return;
//        }

        try {
            tarDatabaseManager.commit();
        } catch (SQLException e) {
            throw new WolfException(String.format("SQLException error:%s", e.getMessage()));
        }
    }

    private void init(EtlDefineModel etlDefineModel) {
        this.needCommit = false;
        this.defind_id = etlDefineModel.getId();
        this.etlDefineModel = etlDefineModel;
        this.srcDataSourceId = etlDefineModel.getSource_data_source_id();
        this.targetDataSourceId = etlDefineModel.getTarget_data_source_id();
    }

    private void reset() {
        values.clear();
        fields.clear();
        orderbys.clear();
        tabs.clear();
        selects = new StringBuffer();
        tables = new StringBuffer();
        joins = new StringBuffer();
        filters = new StringBuffer();
    }

    public EtlDataSourceModel findDataSourceDefineModel(Long dataSourceId) {
        EtlDataSourceModel model = dataSourceDefineService.findById(dataSourceId);
        if (model == null) {
            throw new WolfException(String.format("ETL[id=>%s] can't find any datasource record.", etlDefineModel.getId()));
        }
        return model;
    }

    public String buildSqlOrderby() {
        StringBuffer sb = new StringBuffer();
        sb.append(ORDER_BY_STR);
        EtlTableModel etlTableModel = findEtlTableModel(tabId);
        String alias = getTableAlias(etlTableModel);
        List<EtlTableOrderbyModel> models = findOrderby(tabId);
        if (!CollectionUtils.isEmpty(models)) {
            int i = 0;
            for (EtlTableOrderbyModel model : models) {
                sb.append(String.format(PREFFIX_FIELD, alias, model.getColumn_name()));
                if (i < models.size() - 1) {
                    sb.append(COMMA);
                }
            }
            sb.append(etlTableModel.getIs_asc());
            KeyValuePair kvp = new KeyValuePair(String.valueOf(tabId), sb.toString());
            orderbys.add(kvp);
            return sb.toString();
        }
        return null;
    }

    public void convert(EtlTransformDefineModel defineModel, List<EtlTransformRuleModel> ruleModels, String tableAlias) {
        AbstractConvertRule.RetInfo retInfo = testConvert(ruleModels,
                defineModel.getSrc_column_name(),
                defineModel.getTarget_column_name());

        if (StringUtils.isNotEmpty(retInfo.getTargetInfo())) {
            fields.add(retInfo.getTargetInfo());
        }

        if (IS_CONST.NO.isEquals(defineModel.getIs_const())) {
            fields.add(String.format(PREFFIX_FIELD, tableAlias, defineModel.getSrc_column_name()));

            if (!StringUtils.isEmpty(retInfo.getSrouceInfo())) {
                values.add(retInfo.getSrouceInfo());
            }
        }

        String constValue = defineModel.getConst_value();
        if (IS_CONST.YES.isEquals(defineModel.getIs_const())) {
            if (!isNumberType(tabId, defineModel.getTarget_column_name())) {
                constValue = String.format(SQL_STR_VALUE, constValue);

                if (!StringUtils.isEmpty(constValue)) {
                    values.add(constValue);
                }
            }
        }
    }

    public EtlTransformRuleTypeModel findCheckRuleTypeModel(Long typeId) {
        return ruleTypeService.findCheckRuleTypeModel(typeId);
    }


    public List<EtlTransformRuleModel> findEtlTransformRuleModelsById(Long transDefId) {
        return ruleService.findTransRuleByTransId(transDefId);
    }

    public BaseResult testCheck(List<EtlTransformRuleModel> ruleModels, String srcColumn, String targetColumn) {
        BaseResult result = BaseResult.getSuccess();

//        String srcDataType = findDbDataType(srcColumn);
//        String targetDataType = findDbDataType(targetColumn);
        //....handle
        for (EtlTransformRuleModel ruleModel : ruleModels) {
            EtlTransformRuleTypeModel typeModel = findCheckRuleTypeModel(ruleModel.getType_id());
            if (typeModel == null) {

                return BaseResult.getFailure(String.format("This situation[trans define id=>%s] can't find any transform rule type.",
                        ruleModel.getTrans_def_id()), null);

            }

            AbstractCheckRule rule = AbstractCheckRule.getInstance(CHECK_RULE_TYPE.valueOf(typeModel.getCode()), this);
//            result = rule.check(ruleModel.getTransform_cxt(), srcColumn, targetColumn);
            if (!result.isSucc()) {
                return result;
            }
        }
        return result;
    }

    public boolean testLegalConvertFormat(String cxt) {
        if (StringUtils.isEmpty(cxt)) {
            return false;
        }

        if (cxt.trim().indexOf("%s") == -1) {
            return false;
        }
        return true;
    }

    private AbstractConvertRule.RetInfo testConvert(List<EtlTransformRuleModel> ruleModels, String srcColumn, String targetColumn) {
        if (CollectionUtils.isEmpty(ruleModels)) {
            return new AbstractConvertRule.RetInfo(srcColumn, targetColumn);
        }

        for (EtlTransformRuleModel ruleModel : ruleModels) {
            EtlTransformRuleTypeModel typeModel = findCheckRuleTypeModel(ruleModel.getType_id());
            if (typeModel == null) {
                continue;
            }

            String cxt = ruleModel.getTransform_cxt();
            if (testLegalConvertFormat(cxt)) {
                continue;
            }

            if (typeModel.getIs_check()) {
                AbstractConvertRule convertRule = AbstractConvertRule.getInstance(CONVERT_RULE_TYPE.valueOf(typeModel.getCode()), this);

                AbstractConvertRule.RetInfo retInfo = convertRule.convert(ruleModel.getTransform_cxt(), srcColumn, targetColumn);

                if (APPLICABLE_SCOPE.Both.isEquals(typeModel.getApplicable_scope())) {
                    srcColumn = retInfo.getSrouceInfo();
                    targetColumn = retInfo.getTargetInfo();
                }

                if (APPLICABLE_SCOPE.Source.isEquals(typeModel.getApplicable_scope())) {
                    srcColumn = retInfo.getSrouceInfo();
                }

                if (APPLICABLE_SCOPE.Target.isEquals(typeModel.getApplicable_scope())) {
                    targetColumn = retInfo.getTargetInfo();
                }
            }
        }

        return new AbstractConvertRule.RetInfo(srcColumn, targetColumn);
    }

    public String buildSql(EtlTableModel master) {
        tabId = master.getId();
        String selectPart = buildSelectPart(master.getAlias());

        // 查询关联表
        List<EtlTableModel> etlTableModels = findAllEtlExtTableModels(tabId);

        //判断关联表是否存在
        etlTableModels.forEach((etlTableModel -> testExistsTable(etlTableModel.getId())));

        selects.append(selectPart);

        tables.append(String.format(TABLE_ALIAS, master.getName(), master.getAlias()));
        if (CollectionUtils.isEmpty(etlTableModels)) {
            return tables.toString();
        }

        if (!CollectionUtils.isEmpty(etlTableModels)) {
            tables.append(COMMA);
        }

        //filter part
        etlTableModels.addAll(buildFilterParts());
        for (String filterStr : childrenFilters) {
            filters.append(filterStr);
        }

        int pos = 0;
        for (EtlTableModel model : etlTableModels) {
            String alias = getTableAlias(model);
//            selectPart = buildSelectPart(alias);
//            selects.append(selectPart);
            EtlTableRelModel relModel = findEtlTableExtRelModel(master.getId(), model.getId());
            if (relModel == null) {
                throw new WolfException(String.format("Can't find any rel table info[master table:%s,rel table:%s]",
                        relModel.getMaster_table_id(),
                        relModel.getSlave_table_id()));
            }

            testExistsTable(relModel.getMaster_table_id(), relModel.getMaster_column());
            testExistsTable(relModel.getSlave_table_id(), relModel.getSlave_column());

            String leftStr = buildJoinPart(alias, relModel.getMaster_column());
            String rightStr = buildJoinPart(alias, relModel.getSlave_column());

            joins.append(String.format(TABLE_JOIN_STR, leftStr, rightStr));
            if (pos < etlTableModels.size()) {
                selects.append(COMMA);
                tables.append(COMMA);
            }
            pos++;
        }

        String sql = String.format(SELECT_TABLE_JOIN,
                selects.toString(),
                tables.toString(),
                joins.toString(),
                filters.toString(),
                buildSqlOrderby()
        );
        return sql.replace(ASTERISK, REPLACE_HOLDER);
    }

    private List<EtlTableModel> buildFilterParts() {
        List<EtlTableModel> etlTableModels = new ArrayList<EtlTableModel>();
        Map<KeyValuePair, EtlTableFilterModel> map = buildCondition(tabId);
        Iterator<Map.Entry<KeyValuePair, EtlTableFilterModel>> it = map.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<KeyValuePair, EtlTableFilterModel> entry = it.next();
            KeyValuePair kvp = entry.getKey();
            EtlTableFilterModel filterModel = entry.getValue();
            Long tableId = filterModel.getEtl_table_id();
            if (tableId == Long.parseLong(kvp.getKey())) {
                EtlTableModel etlTableModel = findEtlTableModel(tableId);
                if (!testMasterTableFilter(tableId)) {
                    etlTableModels.add(etlTableModel);
                    continue;
                }

                tables.append(String.format(TABLE_ALIAS, etlTableModel.getName(), etlTableModel.getAlias()));

                String alias = etlTableModel.getName();
                if (StringUtils.isEmpty(etlTableModel.getAlias())) {
                    alias = etlTableModel.getAlias().trim();
                }

                filterModel.check();
                String conditionStr = buildFormula(tableId,
                        alias,
                        filterModel.getColumn_name(),
                        filterModel.getColumn_value1(),
                        filterModel.getColumn_value2());
//                String conditionStr = String.format(TABLE_JOIN_STR,
//                        String.format(PREFFIX_FIELD, alias, filterModel.getColumn_name()),
//                        filterModel.getColumn_value1());
                filters.append(conditionStr);

                if (it.hasNext()) {
                    tables.append(COMMA);
                    filters.append(COMMA);
                }
            }
        }
        return etlTableModels;
    }

    private String buildFormula(Long tabId, String alias, String columnName, String sqlFormula, String... values) {
        if (!StringUtils.isEmpty(alias)) {
            columnName = String.format(PREFFIX_FIELD, alias, columnName);
        }

        AbstractFilter filter = AbstractFilter.getInstance(this, SQL_FORMULA.valueOf(sqlFormula));
        if (filter == null) {
            throw new WolfException(String.format("Can't find any Filter class " +
                            "by sqlFormula[tab id=>%s,column=>%s,formula=>%s].",
                    tabId,
                    columnName,
                    sqlFormula));
        }

        BaseResult result = filter.filter(columnName, values);
        if (!result.isSucc()) {
            throw new WolfException(result.getMessage());
        }

        return result.getMessage();
    }

    private String buildJoinPart(String alias, String columnName) {
        return String.format(PREFFIX_FIELD, alias, columnName);
    }

    private String buildSelectPart(String alias) {
        return String.format(SELECT_PART_STR, alias);
    }

    public String getTableAlias(EtlTableModel model) {
        if (StringUtils.isEmpty(model.getName())) {
            throw new WolfException(String.format("ETL table info[id=>%] it's table name is empty.", model.getId()));
        }

        String alias = model.getName();
        if (!StringUtils.isEmpty(model.getAlias())) {
            alias = model.getAlias();
        }
        return alias;
    }

    private DataSource getDataSourceById(Long dataSourceId) {
        return dynamicService.getById(dataSourceId + "");
    }

    private DatabaseManager getDatabaseManager(Long dataSourceId) {
        try {
            return DatabaseManager.getInstance(getDataSourceById(dataSourceId));
        } catch (Exception e) {
            e.printStackTrace();
        }
        throw new WolfException(String.format("Can't build Database Manager[data source id=>%s] failure!", dataSourceId));
    }

    public Database getDataBaseInstance(Long dataSourceId) {
        try {
            if (databaseMap.get(dataSourceId) != null) {
                return databaseMap.get(dataSourceId);
            }
            Database database = getDatabaseManager(dataSourceId).newDatabase(null, null);
            databaseMap.put(dataSourceId, database);
            return database;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        throw new WolfException(String.format("Can't build database[data source id:%s", dataSourceId));
    }

    public Database getTable(Long dataSourceId, Database database, String tableName) {
        try {
            database = DatabaseManager.getTable(getDataSourceById(dataSourceId), database, tableName);
            databaseMap.put(dataSourceId, database);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return database;
    }

    public void buildDatabaseManager() {
        sourceSqls.clear();
        targetSqls.clear();
        writeBackSqls.clear();
        if (srcDatabaseManager == null) {
            srcDatabaseManager = getDatabaseManager(srcDataSourceId);
            try {
                srcDatabaseManager.init();
            } catch (SQLException e) {
                throw new WolfException(String.format("init error:%s", e.getClass()));
            }
        }

        if (srcDataSourceId == targetDataSourceId) {
            tarDatabaseManager = srcDatabaseManager;
            return;
        }

        if (tarDatabaseManager == null) {
            tarDatabaseManager = getDatabaseManager(targetDataSourceId);
            try {
                tarDatabaseManager.init();
            } catch (SQLException e) {
                throw new WolfException(String.format("init error:%s", e.getClass()));
            }
        }
    }

    public void appendSourceSql(String sql) {
        if (StringUtils.isEmpty(sql)) {
            return;
        }
        this.sourceSqls.add(sql);
    }

    public void appendTargetSql(String sql) {
        if (StringUtils.isEmpty(sql)) {
            return;
        }
        this.targetSqls.add(sql);
    }

    public void appendWriteBackSql(String sql) {
        if (StringUtils.isEmpty(sql)) {
            return;
        }
        this.writeBackSqls.add(sql);
    }

    public BaseResult doSrcExecute() {
        return srcDatabaseManager.execute(sourceSqls);
    }

    public BaseResult doTarExecute() {
        return tarDatabaseManager.execute(targetSqls);
    }

    public BaseResult doSrcWriteBack() {
        return srcDatabaseManager.execute(writeBackSqls);
    }

    public List<Map<String, ColumnDesc>> queryResults(String sql) {
        try {
            DatabaseManager dm = getDatabaseManager(srcDataSourceId);
            return dm.queryResults(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }


    public DIALECT getDataSourceDialet(Long dataSourceId) {
        EtlDataSourceModel dataSourceDefineModel = findDataSourceDefineModel(dataSourceId);
        if (DIALECT.UNKNOWN.isEquals(dataSourceDefineModel.getDialect())) {
            throw new WolfException(String.format("The Dialect of datasource[id=>%s] is unknow.", dataSourceId));
        }

        return DIALECT.value(dataSourceDefineModel.getDialect());
    }

    public void init(Long dataSourceId) {
        Database database = getDataBaseInstance(dataSourceId);
        logger.info(JSONUtils.toJSONString(database));

    }

    public void error(Long defind_id, Long table_id, String column_name, String source_id, String error_code, String error_msg) {
        EtlErrorLogModel model = new EtlErrorLogModel();
        model.setDefine_id(defind_id);
        model.setTable_id(table_id);
        model.setColumn_name(column_name);
        model.setSource_id(source_id);
        model.setError_code(error_code);
        model.setError_msg(error_msg);

        etlErrorLogService.insert(model);
    }

    public Boolean getTest() {
        return isTest;
    }

    public void setTest(Boolean test) {
        isTest = test;
    }
}